package com.wwlwxg.redis;

import java.util.AbstractMap.SimpleEntry;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.StreamEntry;
import redis.clients.jedis.StreamEntryID;

public class StreamConsumeAsync {

	static final String host = "10.92.35.24";
	static final int port = 6300;

	static Jedis jedis = new Jedis(host, port);

	public static void main(String[] args) throws Exception{
		// Start from 0. For subsequent queries, read from last id + 1
		jedis.clientSetname("StreamConsumeAsync");
		String lastStreamDataId = "0-0";
		int count = 1000;
		long waitTimeInMillis = 5000;
		try {
			// Read new data asynchronously in a loop
			while(true) {
				List next = getNext("MyStream", lastStreamDataId,
						count, waitTimeInMillis);
				if(next != null) {
					List<StreamEntry> stList = getStreamEntries(next);
					if(stList != null) {
						// Process data here
						for(int j=0; j<stList.size(); j++) {
							StreamEntry streamData = stList.get(j); // Read the fields (key-value pairs) of data stream
							Map<String, String> fields = streamData.getFields(); // Read subsequent data from last id + 1
							lastStreamDataId = streamData.getID().getTime() + "-"
							+(streamData.getID().getSequence()+1);
							System.out.println(stList.get(j));
							System.out.println(lastStreamDataId);
						}
					} else {
						System.out.println("No new data in the stream");
					}
				}
			}
		} finally {
			jedis.close();
		}
	}

	// Read the next set of data from the stream
	private static List getNext(String streamId, String lastId, int count, long waitTimeInMillis) throws Exception{
		HashMap<String, StreamEntryID> map = new HashMap();
		String readFrom = lastId;
		map.put(streamId, new StreamEntryID(readFrom));
		List list = jedis.xread(count, waitTimeInMillis,
				(Map.Entry<String, StreamEntryID>)
						map.entrySet().toArray()[0]);
		return list;
	}

	// Read stream entries
	// Assumes streamList has only one stream
	private static List<StreamEntry> getStreamEntries(List streamList) throws Exception{
		if(streamList.size()>0) {
			SimpleEntry stEntry = (SimpleEntry)streamList.get(0);
			return (List<StreamEntry>) stEntry.getValue();
		} return null;
	}
}
